package com.hung.trade.goods.mq.processer;

import com.alibaba.fastjson.JSON;
import com.hung.trade.common.constants.TradeEnums;
import com.hung.trade.common.protocol.OrderCancelMessage;
import com.hung.trade.common.protocol.goods.AddGoodsNumberReq;
import com.hung.trade.common.rocketmq.IMessageProcesser;
import com.hung.trade.dao.mapper.TradeMqConsumerLogMapper;
import com.hung.trade.dao.mapper.TradeOrderMapper;
import com.hung.trade.dao.pojo.TradeMqConsumerLog;
import com.hung.trade.dao.pojo.TradeMqConsumerLogExample;
import com.hung.trade.goods.service.IGoodsService;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;

/**
 * @author hungki
 * @createTime 2018-10-09 下午 2:18
 */
public class GoodsOrderCancelProcesser implements IMessageProcesser {
    private static final Logger logger = LoggerFactory.getLogger(GoodsOrderCancelProcesser.class);
    @Autowired
    private TradeOrderMapper tradeOrderMapper;
    @Autowired
    private TradeMqConsumerLogMapper tradeMqConsumerLogMapper;
    @Autowired
    private IGoodsService iGoodsService;

    private String groupName="goods_order_topic_cancel_group";

    private  TradeMqConsumerLog tradeMqConsumerLog;
    @Override
    public boolean heandlerMessage(MessageExt messageExt) {
        try {
            String body =new String(messageExt.getBody(),"UTF-8");
            String topic = messageExt.getTopic();
            String msgId = messageExt.getMsgId();
            String keys = messageExt.getKeys();
            String tags = messageExt.getTags();

            logger.info(String.format("recive message [ body = %s ] ",body));



            TradeMqConsumerLogExample tradeMqConsumerLogExcemple= new TradeMqConsumerLogExample();
            tradeMqConsumerLogExcemple.createCriteria()
                    .andGroupNameEqualTo(groupName)
                    .andMsgTagNotEqualTo(tags)
                    .andMsgKeysEqualTo(keys);

            List<TradeMqConsumerLog> consumerLogList = tradeMqConsumerLogMapper.selectByExample(tradeMqConsumerLogExcemple);

            if(null==consumerLogList || consumerLogList.size()<=0){
                try {
                    tradeMqConsumerLog = new TradeMqConsumerLog();
                    tradeMqConsumerLog.setConsumerTimes(1);
                    tradeMqConsumerLog.setGroupName(groupName);
                    tradeMqConsumerLog.setMsgBody(body);
                    tradeMqConsumerLog.setMsgId(msgId);
                    tradeMqConsumerLog.setMsgKeys(keys);
                    tradeMqConsumerLog.setMsgTag(tags);
                    tradeMqConsumerLog.setConsumerTimes(0);
                    tradeMqConsumerLog.setConsumerStatus(TradeEnums.ConsumerStatusEnums.PROCESSING.getStatusCode());
                    //这里，tags,keys,groupname应该是联合主键的，然而没有，所以不会出现主键冲突
                    tradeMqConsumerLogMapper.insert(tradeMqConsumerLog);

                }catch (Exception e){
                    //表中存在消费记录，直接返回稍后重试
                    return false;
                }
                return true;

            }else{
                tradeMqConsumerLog = consumerLogList.get(0);

                if(tradeMqConsumerLog.getConsumerStatus().equals(TradeEnums.ConsumerStatusEnums.SUCCESS.getClass())){
                    //已经消费成功了，直接返回true不在消费
                    return true;
                }else if(tradeMqConsumerLog.getConsumerStatus().equals(TradeEnums.ConsumerStatusEnums.PROCESSING.getClass())){
                    //当前组内有其他的消费者在消费，返回false，稍后重试
                    return false;
                }else if(tradeMqConsumerLog.getConsumerStatus().equals(TradeEnums.ConsumerStatusEnums.FAIL.getClass())) {
                    if (tradeMqConsumerLog.getConsumerTimes() >= 5) {
                        //处理超过三次都失败了，返回true，这条消息入库等待人工处理 TODO 需要入库
                        return true;
                    }

                    tradeMqConsumerLog.setConsumerStatus(TradeEnums.ConsumerStatusEnums.PROCESSING.getStatusCode());
                    tradeMqConsumerLog.setConsumerTimes(tradeMqConsumerLog.getConsumerTimes()+1);
                    TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
                    example.createCriteria().andMsgTagNotEqualTo(tradeMqConsumerLog.getMsgTag())
                            .andMsgKeysEqualTo(tradeMqConsumerLog.getMsgKeys())
                            .andGroupNameEqualTo(groupName)
                            .andConsumerTimesEqualTo(tradeMqConsumerLog.getConsumerTimes())
                            .andConsumerStatusEqualTo(TradeEnums.ConsumerStatusEnums.FAIL.getStatusCode());
                    //只有都满足example里面的条件是才允许更新，就是采用乐观锁的方式进行更新
                    int i =tradeMqConsumerLogMapper.updateByExample(tradeMqConsumerLog,example);
                    if(i<=0){
                        logger.info("遇到了并发更新状态，返回false，一会重试");
                        return false;
                    }else{
                        return true;
                    }

                }

            }


            //业务处理
            OrderCancelMessage orderCancelMessage = JSON.parseObject(body,OrderCancelMessage.class);
            AddGoodsNumberReq req=new AddGoodsNumberReq();
            req.setOrderId(orderCancelMessage.getOrderId());
            req.setGoodsId(orderCancelMessage.getGoodsId());
            req.setGoodsNumber(orderCancelMessage.getGoodsNumber());
            iGoodsService.addGoodsNumber(req);

            //更新消费记录的状态为处理完成

            tradeMqConsumerLog.setConsumerStatus(TradeEnums.ConsumerStatusEnums.SUCCESS.getStatusCode());
            tradeMqConsumerLog.setConsumerTimes(tradeMqConsumerLog.getConsumerTimes()+1);
            TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
            example.createCriteria().andMsgTagNotEqualTo(tags)
                    .andMsgKeysEqualTo(keys)
                    .andGroupNameEqualTo(groupName);
            tradeMqConsumerLogMapper.updateByExample(tradeMqConsumerLog,example);

            return  true;

        } catch (Exception e) {

            //更新消费记录的状态为处理完成
            tradeMqConsumerLog.setConsumerStatus(TradeEnums.ConsumerStatusEnums.FAIL.getStatusCode());
            tradeMqConsumerLog.setConsumerTimes(tradeMqConsumerLog.getConsumerTimes()+1);
            TradeMqConsumerLogExample example = new TradeMqConsumerLogExample();
            example.createCriteria().andMsgTagNotEqualTo(tradeMqConsumerLog.getMsgTag())
                    .andMsgKeysEqualTo(tradeMqConsumerLog.getMsgKeys())
                    .andGroupNameEqualTo(groupName);
            tradeMqConsumerLogMapper.updateByExample(tradeMqConsumerLog,example);
            return false;
        }

    }
}
